package cn.zwx.common.canal.service.impl;

import cn.zwx.common.canal.bo.CanalEntryBO;
import cn.zwx.common.canal.config.CanalPropertiesConfig;
import cn.zwx.common.canal.event.AlterAbstractCanalEvent;
import cn.zwx.common.canal.event.DeleteAbstractCanalEvent;
import cn.zwx.common.canal.event.InsertAbstractCanalEvent;
import cn.zwx.common.canal.event.UpdateAbstractCanalEvent;
import cn.zwx.common.canal.service.CanalHandlerService;
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;

/**
 * @author zhangwenxue
 */
@Component
@EnableConfigurationProperties(CanalPropertiesConfig.class)
public class CanalHandlerServiceImpl implements CanalHandlerService {

    public static final Logger logger = LoggerFactory.getLogger(CanalHandlerServiceImpl.class);
    @Resource
    private ApplicationContext applicationContext;

    @Resource
    public CanalPropertiesConfig propertiesConfig;


    @Override

    public void processBinlog(String destination, CanalConnector canalConnector) {
        Message message = canalConnector.getWithoutAck(propertiesConfig.getBatchSize());
        long batchId = message.getId();
        final List<CanalPropertiesConfig.CanalInstance> serverAddrList = propertiesConfig.getInstances();
        final ConcurrentMap<String, String> instanceMap = serverAddrList.stream()
                .collect(Collectors.toConcurrentMap(CanalPropertiesConfig.CanalInstance::getDestination,
                        CanalPropertiesConfig.CanalInstance::getTopic, (key1, key2) -> key2));
        try {
            List<CanalEntry.Entry> entries = message.getEntries();
            if (!(batchId != -1 && entries.size() > 0)) {
                return;
            }
            for (CanalEntry.Entry entry : entries) {
                //  2.发布对应的事件
                logger.info("publishCanalEvent=" + entry.getEntryType());
                if (!entry.getEntryType().name().equals(CanalEntry.EntryType.ROWDATA.name())) {
                    continue;
                }
                // 3.获取对象实例的Topic
                final String topic = instanceMap.get(destination);
                if (StringUtils.isEmpty(topic)){
                    logger.error("processBinlog not found topic:{}", JSON.toJSONString(entry));
                    continue;
                }
                // 发布监听事件
                this.publishCanalEvent(new CanalEntryBO(entry,destination,topic));
            }
            canalConnector.ack(batchId);
        } catch (Exception e) {
            logger.error("发送监听事件失败！batchId回滚,batchId=,原因：{}", batchId, e);
            canalConnector.rollback(batchId);
        }
    }

    private void publishCanalEvent(CanalEntryBO canalEntryBO) {
        CanalEntry.Entry entry = canalEntryBO.getEntry();
        CanalEntry.EventType eventType = entry.getHeader().getEventType();
        switch (eventType) {
            case INSERT:
                applicationContext.publishEvent(new InsertAbstractCanalEvent(canalEntryBO));
                break;
            case UPDATE:
                applicationContext.publishEvent(new UpdateAbstractCanalEvent(canalEntryBO));
                break;
            case DELETE:
                applicationContext.publishEvent(new DeleteAbstractCanalEvent(canalEntryBO));
                break;
            default:
                applicationContext.publishEvent(new AlterAbstractCanalEvent(canalEntryBO));
        }
    }
}
